package nil.ed.simulator.request.event.bus;

import com.google.inject.Inject;
import nil.ed.simulator.request.enums.events.EventMode;
import nil.ed.simulator.request.event.Event;
import nil.ed.simulator.request.event.EventBusAware;
import nil.ed.simulator.request.event.annotation.DependOns;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 通用事件总线.
 * @author lidelin.
 */
public class StandardEventBus implements EventBus {

    private static final int DEFAULT_ASYNC_CORE_SIZE = 2;
    private static final int DEFAULT_ASYNC_MAX_SIZE = 10;

    private final Object ASYNC_EXECUTOR_INIT_LOCK = new Object();

    /**
     * 监听器.
     */
    private final List<EventListener> listeners = new LinkedList<>();

    /**
     * 异步执行器.
     */
    private ExecutorService asyncExecutor;

    @Inject
    public StandardEventBus(Set<EventListener> listeners) {
        this.listeners.addAll(listeners);
        listeners.stream().filter(e -> EventBusAware.class.isAssignableFrom(e.getClass()))
                .map(EventBusAware.class::cast)
                .forEach(e -> e.setEventBus(this));
        getSupportedListeners(null);
    }

    @Override
    public synchronized void publish(Event...events) {
        for (Event event : events) {
            List<EventListener> eventListeners = getSupportedListeners(event);
            if (event.getEventMode() == EventMode.SYNC) {
                eventListeners.forEach(eventListener -> eventListener.onEvent(event));
            } else {
                eventListeners.forEach(eventListener -> {
                    getAsyncExecutorInternal().execute(() -> eventListener.onEvent(event));
                });
            }
        }
    }

    @Override
    public synchronized void subscribe(EventListener listener) {
        listeners.add(listener);
    }

    public List<EventListener> getSupportedListeners(Event event) {
        if (listeners.isEmpty()) {
            return Collections.emptyList();
        }

        List<EventListener> candidateEventListeners = listeners.stream().filter(listener -> event == null || listener.isSupport(event)).collect(Collectors.toList());
        List<EventListener> result = new LinkedList<>();
        Set<Class<?>> resultClazzSet = new HashSet<>();
        while (result.size() != candidateEventListeners.size()) {
            int lastSize = result.size();
            candidateEventListeners.stream().filter(e -> {
                        DependOns dependOns = e.getClass().getAnnotation(DependOns.class);
                        return dependOns == null || Stream.of(dependOns.depends()).allMatch(resultClazzSet::contains);
                    }).forEach(e -> {
                        resultClazzSet.add(e.getClass());
                        result.add(e);
            });
            if (lastSize == result.size()) {
                throw new RuntimeException("Listener Cycle Detection!");
            }
        }
        return result;
    }

    public List<EventListener> getListeners() {
        return listeners;
    }

    public ExecutorService getAsyncExecutor() {
        return asyncExecutor;
    }

    public void setAsyncExecutor(ExecutorService asyncExecutor) {
        this.asyncExecutor = asyncExecutor;
    }

    protected ExecutorService getAsyncExecutorInternal() {
        if (asyncExecutor == null) {
            synchronized (ASYNC_EXECUTOR_INIT_LOCK) {
                if (asyncExecutor == null) {
                    asyncExecutor = new ThreadPoolExecutor(DEFAULT_ASYNC_CORE_SIZE, DEFAULT_ASYNC_MAX_SIZE,
                            10, TimeUnit.MINUTES, new LinkedBlockingQueue<>(100), new BasicThreadFactory.Builder().namingPattern("Event-Async-%s").build(),  new ThreadPoolExecutor.CallerRunsPolicy());
                }
            }
        }
        return asyncExecutor;
    }
}
